package removeDanglingPage;

import java.io.IOException;

import org.apache.hadoop.fs.FileSystem;
import org.apache.hadoop.io.BytesWritable;
import org.apache.hadoop.io.SequenceFile;
import org.apache.hadoop.io.compress.DefaultCodec;
import org.apache.hadoop.io.compress.GzipCodec;
import org.apache.hadoop.mapred.JobConf;
import org.apache.hadoop.mapred.RunningJob;
import org.apache.hadoop.mapred.SequenceFileInputFormat;
import org.apache.hadoop.mapred.SequenceFileOutputFormat;
import org.apache.hadoop.mapred.TextOutputFormat;

import mapred.CombineConfig;
import mapred.InputConfig;
import mapred.Job;
import mapred.MapConfig;
import mapred.OutputConfig;
import mapred.ReduceConfig;

public class RemoveDP_GetReady extends Job{
	
	private String input;
	private String output;
	
	public RemoveDP_GetReady(String input, String output) {
		super("RemoveDanglingPage_GetReady", new String[0]);
		this.input = input;
		this.output = output;
	}

	@Override
	protected CombineConfig getCombineConfig() {
		// TODO Auto-generated method stub
		return CombineConfig.newInstance(null);
	}

	@Override
	protected InputConfig getInputConfig(FileSystem fs) throws IOException {
		// TODO Auto-generated method stub
		return InputConfig.newInstance(SequenceFileInputFormat.class, input);
	}

	@Override
	protected String getLogPath() {
		// TODO Auto-generated method stub
//		return "/user/jwLiu/logs";
		return "/user/jwLiu/CJKPageRank/logs";
	}

	@Override
	protected MapConfig getMapConfig() {
		// TODO Auto-generated method stub
		return MapConfig.newInstance(RemoveDP_GetReady_Mapper.class, 0, BytesWritable.class, BytesWritable.class, DefaultCodec.class);
	}

	@Override
	protected OutputConfig getOutputConfig(FileSystem fs) throws IOException {
		// TODO Auto-generated method stub
		return OutputConfig.newInstance(SequenceFileOutputFormat.class, BytesWritable.class, BytesWritable.class, output+"/RemoveDP_0", GzipCodec.class);
	}

	@Override
	protected ReduceConfig getReduceConfig() {
		// TODO Auto-generated method stub
		return ReduceConfig.newInstance(RemoveDP_GetReady_Reducer.class, 5000);
	}

	@Override
	protected void setUp(JobConf conf, FileSystem fs) throws IOException {
		// TODO Auto-generated method stub
		
		SequenceFileOutputFormat.setOutputCompressionType(conf, SequenceFile.CompressionType.BLOCK);
		conf.setPartitionerClass(PartitionConf.class);
//		conf.setOutputKeyComparatorClass(Compare.class);
		conf.set("dfs.http.address", "10.251.2.1:50070");
//		conf.set("io.sort.mb", "300");
//		conf.set("io.sort.factor", "100");
		conf.set("mapred.child.java.opts", "-Xmx1024m");
	}

	@Override
	protected void tearDown(JobConf conf, FileSystem fs, RunningJob job)
			throws IOException {
		// TODO Auto-generated method stub
		
	}

}
